package com.rocketmq.demo;

import com.rocketmq.constants.MqConstants;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.junit.Test;

/**
 * @author: Patrick Star
 * @description:
 * @Date: 2025-07-27
 */
public class BAsyncTest {

    @Test
    public void asyncProducer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("async-producer-group");
        producer.setNamesrvAddr(MqConstants.NAMESERVER_ADDR);
        producer.start();

        Message message = new Message("asyncTopic", "我是一个异步消息".getBytes());

        producer.send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("发送成功");
            }

            @Override
            public void onException(Throwable e) {
                System.out.println("发送失败" + e.getMessage());
            }
        });

        System.out.println("我先执行");
        System.in.read();
    }

    @Test
    public void asyncConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("async-consumer-group");
        consumer.setNamesrvAddr(MqConstants.NAMESERVER_ADDR);
        consumer.subscribe("asyncTopic", "*");

        consumer.registerMessageListener((MessageListenerConcurrently) (message, context) -> {
            System.out.println("我是一个异步消费者");
            System.out.println(message);
            System.out.println(new String(message.get(0).getBody()));
            System.out.println(context);
            return null;
        });

        consumer.start();
        System.in.read();
    }
}
